package top.aniss.spark.beifeng.spark;

import com.alibaba.fastjson.JSONObject;
import org.apache.log4j.Logger;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.hive.HiveContext;
import scala.Tuple2;
import top.aniss.spark.beifeng.common.Const;
import top.aniss.spark.beifeng.common.MockData;
import top.aniss.spark.beifeng.conf.ConfigurationManager;
import top.aniss.spark.beifeng.dao.DAOFactory;
import top.aniss.spark.beifeng.dao.ISessionAggrStatDAO;
import top.aniss.spark.beifeng.dao.ITaskDAO;
import top.aniss.spark.beifeng.domain.SessionAggrStat;
import top.aniss.spark.beifeng.domain.Task;
import top.aniss.spark.beifeng.util.*;

import java.util.Date;
import java.util.Iterator;

/**
 * 用户访问session分析Spark作业
 * <p>
 * 接收用户创建的分析任务，用户可能指定的条件如下：
 * <p>
 * 1、时间范围：起始日期~结束日期
 * 2、性别：男或女
 * 3、年龄范围
 * 4、职业：多选
 * 5、城市：多选
 * 6、搜索词：多个搜索词，只要某个session中的任何一个action搜索过指定的关键词，那么session就符合条件
 * 7、点击品类：多个品类，只要某个session中的任何一个action点击过某个品类，那么session就符合条件
 * <p>
 * 我们的spark作业如何接受用户创建的任务？
 * <p>
 * J2EE平台在接收用户创建任务的请求之后，会将任务信息插入MySQL的task表中，任务参数以JSON格式封装在task_param
 * 字段中
 * <p>
 * 接着J2EE平台会执行我们的spark-submit shell脚本，并将taskid作为参数传递给spark-submit shell脚本
 * spark-submit shell脚本，在执行时，是可以接收参数的，并且会将接收的参数，传递给Spark作业的main函数
 * 参数就封装在main函数的args数组中
 * <p>
 * 这是spark本身提供的特性
 *
 * @author Andre Wei
 * email aniss.wei@gmail.com
 * create 2018-07-11 下午7:13
 */
public class UserVisitSessionAnalyzeSpark {
    private static Logger logger = Logger.getLogger(UserVisitSessionAnalyzeSpark.class);

    public static void main(String[] args) {
        // args = new String[]{"1"};
        // 构建Spark上下文
        SparkConf conf = new SparkConf()
                .setAppName(Const.SPARK.APP_NAME)
                .setMaster("local[2]");

        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = getSQLContext(sc.sc());


        /*
         * 生成模拟数据, 三个临时表
         * user_visit_action
         * user_info
         * product_info
         */
        mockData(sc, sqlContext);

        // 创建需要的DAO组件
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();

        Long taskId = ParamUtils.getTaskIdFromArgs(args, Const.SPARK.TASK_TYPE);
        if (taskId == null) {
            logger.error("无法获取任务ID");
            System.exit(1);
        }
        Task task = taskDAO.findById(taskId);

        // 获取任务参数, 并转换成json格式
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());

        // 通过任务参数获取RDD
        JavaRDD<Row> actionRDD = getActionRDDByDateRange(sqlContext, taskParam);

        JavaPairRDD<String, String> sessionId2AggrInfoRDD = aggregateBySession(sqlContext, actionRDD);

        // TEST BEGIN
        System.out.println(sessionId2AggrInfoRDD.count());

        for (Tuple2<String, String> tuple : sessionId2AggrInfoRDD.take(10)) {
            System.out.println(tuple._2);
        }
        // TEST END

        // Filter and refactor at the same time
        final Accumulator<String> sessionAggrStatAccumulator = sc.accumulator("", new SessionAggrStatAccumulator());

        JavaPairRDD<String, String> filteredSessionId2AggrInfoRDD = filterSessionAndAggrStat(
                sessionId2AggrInfoRDD, taskParam, sessionAggrStatAccumulator);

        // Calculate the proportion of each session step and visit length in total session


        /*
         * 对于Accumulator这种分布式累加计算的变量的使用，有一个重要说明
         *
         * 从Accumulator中，获取数据，插入数据库的时候，一定要，一定要，是在有某一个action操作以后
         * 再进行。。。
         *
         * 如果没有action的话，那么整个程序根本不会运行。。。
         *
         * 是不是在calculateAndPersisitAggrStat方法之后，运行一个action操作，比如count、take
         * 不对！！！
         *
         * 必须把能够触发job执行的操作，放在最终写入MySQL方法之前
         *
         * 计算出来的结果，在J2EE中，是怎么显示的，是用两张柱状图显示
         */
        System.out.println(filteredSessionId2AggrInfoRDD.count());

        calculateAndPersist(sessionAggrStatAccumulator.value(), taskId);


        // TEST BEGIN
        System.out.println(filteredSessionId2AggrInfoRDD.count());
        for (Tuple2<String, String> tuple : filteredSessionId2AggrInfoRDD.take(10)) {
            System.out.println(tuple._2);
        }
        // TEST END

        /*
         * session聚合统计（统计出访问时长和访问步长，各个区间的session数量占总session数量的比例）
         *
         * 如果不进行重构，直接来实现，思路：
         * 1、actionRDD，映射成<sessionid,Row>的格式
         * 2、按sessionid聚合，计算出每个session的访问时长和访问步长，生成一个新的RDD
         * 3、遍历新生成的RDD，将每个session的访问时长和访问步长，去更新自定义Accumulator中的对应的值
         * 4、使用自定义Accumulator中的统计值，去计算各个区间的比例
         * 5、将最后计算出来的结果，写入MySQL对应的表中
         *
         * 普通实现思路的问题：
         * 1、为什么还要用actionRDD，去映射？其实我们之前在session聚合的时候，映射已经做过了。多此一举
         * 2、是不是一定要，为了session的聚合这个功能，单独去遍历一遍session？其实没有必要，已经有session数据
         * 		之前过滤session的时候，其实，就相当于，是在遍历session，那么这里就没有必要再过滤一遍了
         *
         * 重构实现思路：
         * 1、不要去生成任何新的RDD（处理上亿的数据）
         * 2、不要去单独遍历一遍session的数据（处理上千万的数据）
         * 3、可以在进行session聚合的时候，就直接计算出来每个session的访问时长和访问步长
         * 4、在进行过滤的时候，本来就要遍历所有的聚合session信息，此时，就可以在某个session通过筛选条件后
         * 		将其访问时长和访问步长，累加到自定义的Accumulator上面去
         * 5、就是两种截然不同的思考方式，和实现方式，在面对上亿，上千万数据的时候，甚至可以节省时间长达
         * 		半个小时，或者数个小时
         *
         * 开发Spark大型复杂项目的一些经验准则：
         * 1、尽量少生成RDD
         * 2、尽量少对RDD进行算子操作，如果有可能，尽量在一个算子里面，实现多个需要做的功能
         * 3、尽量少对RDD进行shuffle算子操作，比如groupByKey、reduceByKey、sortByKey（map、mapToPair）
         * 		shuffle操作，会导致大量的磁盘读写，严重降低性能
         * 		有shuffle的算子，和没有shuffle的算子，甚至性能，会达到几十分钟，甚至数个小时的差别
         * 		有shfufle的算子，很容易导致数据倾斜，一旦数据倾斜，简直就是性能杀手（完整的解决方案）
         * 4、无论做什么功能，性能第一
         * 		在传统的J2EE或者.NET后者PHP，软件/系统/网站开发中，我认为是架构和可维护性，可扩展性的重要
         * 		程度，远远高于了性能，大量的分布式的架构，设计模式，代码的划分，类的划分（高并发网站除外）
         *
         * 		在大数据项目中，比如MapReduce、Hive、Spark、Storm，我认为性能的重要程度，远远大于一些代码
         * 		的规范，和设计模式，代码的划分，类的划分；大数据，大数据，最重要的，就是性能
         * 		主要就是因为大数据以及大数据项目的特点，决定了，大数据的程序和项目的速度，都比较慢
         * 		如果不优先考虑性能的话，会导致一个大数据处理程序运行时间长度数个小时，甚至数十个小时
         * 		此时，对于用户体验，简直就是一场灾难
         *
         * 		所以，推荐大数据项目，在开发和代码的架构中，优先考虑性能；其次考虑功能代码的划分、解耦合
         *
         * 		我们如果采用第一种实现方案，那么其实就是代码划分（解耦合、可维护）优先，设计优先
         * 		如果采用第二种方案，那么其实就是性能优先
         *
         * 		讲了这么多，其实大家不要以为我是在岔开话题，大家不要觉得项目的课程，就是单纯的项目本身以及
         * 		代码coding最重要，其实项目，我觉得，最重要的，除了技术本身和项目经验以外；非常重要的一点，就是
         * 		积累了，处理各种问题的经验
         *
         */


        // 关闭Spark上下文
        sc.close();
    }


    /**
     * 如果是在本地运行, 就生成SQLContext
     * 如果是在服务器, 就使用HiveContext
     *
     * @param sc sparkContext
     * @return an instance of SQLContext
     */
    private static SQLContext getSQLContext(SparkContext sc) {
        boolean isLocal = ConfigurationManager.getBoolean(Const.SPARK.LOCAL);
        if (isLocal) {
            return new SQLContext(sc);
        } else {
            return new HiveContext(sc);
        }
    }


    /**
     * 只有本地数据才生成模拟数据
     *
     * @param sc sparkContext
     * @param sqlContext sqlContext
     */
    private static void mockData(JavaSparkContext sc, SQLContext sqlContext) {
        boolean isLocal = ConfigurationManager.getBoolean(Const.SPARK.LOCAL);
        if (isLocal) {
            MockData.mock(sc, sqlContext);
        }
    }

    /**
     * 获取指定日期范围内的用户访问数据
     *
     * @param sqlContext SQLContext
     * @param taskParams the task params
     * @return an action RDD
     */
    private static JavaRDD<Row> getActionRDDByDateRange(SQLContext sqlContext, JSONObject taskParams) {
        String startDate = ParamUtils.getParam(taskParams, Const.SPARK.PARAM_START_DATE);
        String endDate = ParamUtils.getParam(taskParams, Const.SPARK.PARAM_END_DATE);

        String sql = String.format(
                "SELECT * FROM user_visit_action where date>='%s' and date<='%s'", startDate, endDate);

        DataFrame actionDF = sqlContext.sql(sql);

        return actionDF.javaRDD();
    }


    /**
     * 对行为数据按session粒度进行聚合
     *
     * @param actionRDD 行为数据RDD
     * @return session粒度聚合数据
     */
    private static JavaPairRDD<String, String> aggregateBySession(
            SQLContext sqlContext, final JavaRDD<Row> actionRDD) {
        // actionRDD中的一个row就是一行用户行为记录, 比如一次点击或者一次搜索
        // 我们现在需要将这个Row映射成<sessionId, Row>的形式

        /*
         * 对于PairFunction
         * 第一个参数, 相当于函数的输入
         * 第二个和第三个参数, 相当于是函数的输出, 为一个Tuple, 分别是Tuple的第一个和第二个值
         */
        JavaPairRDD<String, Row> session2ActionRDD = actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
            @Override
            public Tuple2<String, Row> call(Row row) {
                return new Tuple2<>(row.getString(2), row);
            }
        });
        // 对行为数据按session粒度进行分组, 就是根据sessionId进行GroupBy
        JavaPairRDD<String, Iterable<Row>> sessionId2ActionsRDD = session2ActionRDD.groupByKey();

        // 对每一个session分组进行聚合, 将session中所有的搜索词和品类都聚合起来, 返回 <userId, fullAggrInfo>
        JavaPairRDD<Long, String> userId2PartAggrInfoRDD = sessionId2ActionsRDD.mapToPair(
                new PairFunction<Tuple2<String, Iterable<Row>>, Long, String>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Long, String> call(Tuple2<String, Iterable<Row>> tuple) {
                        String sessionId = tuple._1;
                        Iterator<Row> iterator = tuple._2.iterator();

                        StringBuilder searchKeywordsBuffer = new StringBuilder();
                        StringBuilder clickCategoryIdsBuilder = new StringBuilder();

                        Long userId = null;

                        // the start time and end time of session
                        Date startTime = null;
                        Date endTime = null;

                        // the step length of session
                        long stepLength = 0;

                        // 遍历session中所有的访问行为
                        // 同一个sessionId的所有请求肯定属于同意个userId
                        while (iterator.hasNext()) {
                            // 提取每个访问行为的搜索词字段和点击品类字段
                            Row row = iterator.next();

                            if (userId == null) {
                                userId = row.getLong(1);
                            }
                            String searchKeywords = row.getString(5);

                            Long clickCategoryId;
                            try {
                                clickCategoryId = row.getLong(6);
                            } catch (Exception e) {
                                clickCategoryId = null;
                                logger.error(e);
                            }

                            /*
                             * 并不是每一行访问行为都有searchKeyword和clickCategoryId
                             * 搜索行为才有searchKeyword
                             * 点击品类的行为才有clickCategoryId
                             * 所以, 任何一行行为数据, 都不可能同时存在这两个字段, 必定有一个为null
                             *
                             * 所以我们决定是否将搜索词或者点击品类id拼接到字符串中, 首先要满足不能是null
                             * 其次, 之前的字符串中还没有搜索词或者点击品类id
                             */
                            if (StringUtils.isNotBlank(searchKeywords)) {
                                if (!searchKeywordsBuffer.toString().contains(searchKeywords)) {
                                    searchKeywordsBuffer.append(searchKeywords).append(",");
                                }
                            }
                            if (clickCategoryId != null) {
                                if (!clickCategoryIdsBuilder.toString().contains(String.valueOf(clickCategoryId))) {
                                    clickCategoryIdsBuilder.append(clickCategoryId).append(",");
                                }
                            }

                            // calculate the startTime and endTime
                            String timeStr = row.getString(4);
                            Date actionTime = null;
                            if (StringUtils.isNotBlank(timeStr)) {
                                actionTime = DateUtils.parseTime(timeStr);
                            } else {
                                System.exit(1);
                            }
                            if (startTime == null) {
                                startTime = actionTime;
                            }
                            if (endTime == null) {
                                endTime = actionTime;
                            }
                            if (actionTime != null && actionTime.before(startTime)) {
                                startTime = actionTime;
                            }
                            if (actionTime != null && actionTime.after(endTime)) {
                                endTime = actionTime;
                            }
                            // calculate the step length
                            stepLength++;
                        }
                        String searchKeywords = StringUtils.trimComma(searchKeywordsBuffer.toString());
                        String clickCategoryIds = StringUtils.trimComma(clickCategoryIdsBuilder.toString());

                        // calculate the session visit total time (second)
                        long visitLength = 0;
                        if (startTime != null) {
                            visitLength = (endTime.getTime() - startTime.getTime()) / 1000;
                        }

                        // 返回的数据格式, <sessionId, partAggrInfo>
                        // 但是, 这一步聚合完了以后, 我们还要跟对应的用户信息进行聚合
                        // 如果是跟用户聚合的话, 那么 key 就应该是 userId, 才能跟 <userId, Row> 格式的用户信息进行聚合
                        // 如果我们这里直接返回 <sessionId, partAggrInfo>, 那么就还得做一次 mapToPair 算子

                        // 因此, 这里应该直接返回 <userId, partAggrInfo>
                        // 然后, 跟用户的 join 的时候, 将 partAggrInfo 关联上 userInfo

                        // 聚合数据用什么样的格式进行拼接
                        // 这里同意定义使用 key1=value1|key2=value2
                        String partAggrInfo = Const.SPARK.FIELD_SESSION_ID + "=" + sessionId + "|" +
                                (StringUtils.isNotBlank(searchKeywords)
                                        ? Const.SPARK.FIELD_SEARCH_KEYWORDS + "=" + searchKeywords + "|" : "") +
                                (StringUtils.isNotBlank(clickCategoryIds)
                                        ? Const.SPARK.FIELD_CLICK_CATEGORY_IDS + "=" + clickCategoryIds + "|" : "") +
                                Const.SPARK.FIELD_VISIT_LENGTH + "=" + visitLength + "|" +
                                Const.SPARK.FIELD_STEP_LENGTH + "=" + stepLength;

                        return new Tuple2<>(userId, partAggrInfo);
                    }

                });

        // 查询所有用户数据, 并获取用户信息RDD
        String sql = "SELECT * FROM user_info";
        final JavaRDD<Row> userInfoRDD = sqlContext.sql(sql).javaRDD();

        // 转成<userId, userInfoRow>
        JavaPairRDD<Long, Row> userId2InfoRDD = userInfoRDD.mapToPair(new PairFunction<Row, Long, Row>() {
            @Override
            public Tuple2<Long, Row> call(Row row) {
                return new Tuple2<>(row.getLong(0), row);
            }
        });

        // 将session粒度聚合数据, 与用户信息进行join
        // <userId, <partAggrInfoRDD, InfoRDD>>
        JavaPairRDD<Long, Tuple2<String, Row>> userId2FullInfoRDD = userId2PartAggrInfoRDD.join(userId2InfoRDD);

        // 对Join起来的数据进行拼接, 并且返回<sessionId,fullAggreInfo>格式的数据
        return userId2FullInfoRDD.mapToPair(
                new PairFunction<Tuple2<Long, Tuple2<String, Row>>, String, String>() {
                    @Override
                    public Tuple2<String, String> call(Tuple2<Long, Tuple2<String, Row>> longTuple2Tuple2) {
                        // longTuple2Tuple2 : <userId, <partAggrInfoRDD, InfoRDD>>
                        String partAggrInfo = longTuple2Tuple2._2._1;
                        Row infoRow = longTuple2Tuple2._2._2;

                        String sessionId = StringUtils.getFieldFromConcatString(partAggrInfo, "\\|",
                                Const.SPARK.FIELD_SESSION_ID);

                        int age = infoRow.getInt(3);
                        String professional = infoRow.getString(4);
                        String city = infoRow.getString(5);
                        String sex = infoRow.getString(6);

                        String fullAggrInfo = partAggrInfo + "|" +
                                Const.SPARK.FIELD_AGE + "=" + age + "|" +
                                Const.SPARK.FIELD_PROFESSIONAL + "=" + professional + "|" +
                                Const.SPARK.FIELD_CITY + "=" + city + "|" +
                                Const.SPARK.FIELD_SEX + "=" + sex;
                        return new Tuple2<>(sessionId, fullAggrInfo);
                    }
                });
    }

    /**
     * 过滤session数据， 并进行聚合统计
     *
     * @param sessionId2AggrInfoRDD      sessionId2AggrInfoRDD
     * @param sessionAggrStatAccumulator 累加器
     * @return a pairRDD with <String,String>
     */
    private static JavaPairRDD<String, String> filterSessionAndAggrStat(
            JavaPairRDD<String, String> sessionId2AggrInfoRDD,
            final JSONObject taskParam,
            final Accumulator<String> sessionAggrStatAccumulator) {

        // 为了使用ValidUtils， 所以先将所有的参数拼接成一个字符串
        String startAge = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_START_AGE);
        String endAge = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_END_AGE);
        String professionals = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_PROFESSIONALS);
        String cities = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_CITIES);
        String sex = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_SEX);
        String keywords = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_KEYWORDS);
        String categoryIds = ParamUtils.getParam(taskParam, Const.SPARK.PARAM_CATEGORY_IDS);

        String _parameter = (startAge != null ? Const.SPARK.PARAM_START_AGE + "=" + startAge + "|" : "") +
                (endAge != null ? Const.SPARK.PARAM_END_AGE + "=" + endAge + "|" : "") +
                (professionals != null ? Const.SPARK.PARAM_PROFESSIONALS + "=" + professionals + "|" : "") +
                (cities != null ? Const.SPARK.PARAM_CITIES + "=" + cities + "|" : "") +
                (sex != null ? Const.SPARK.PARAM_SEX + "=" + sex + "|" : "") +
                (keywords != null ? Const.SPARK.PARAM_KEYWORDS + "=" + keywords + "|" : "") +
                (categoryIds != null ? Const.SPARK.PARAM_CATEGORY_IDS + "=" + categoryIds + "|" : "");
        if (StringUtils.isNotBlank(_parameter) && _parameter.endsWith("|")) {
            _parameter = _parameter.substring(0, _parameter.length() - 1);
        }

        final String parameter = _parameter;

        // 根据筛选参数进行过滤
        return sessionId2AggrInfoRDD.filter(new Function<Tuple2<String, String>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, String> tuple) {
                // 首先，从tuple中获取聚合数据
                String aggrInfo = tuple._2;

                // 一次按照筛选条件进行过滤
                // 按照年龄范围进行过滤 (startAge, endAge)
                if (!ValidUtils.between(aggrInfo, Const.SPARK.FIELD_AGE, parameter, Const.SPARK.PARAM_START_AGE,
                        Const.SPARK.PARAM_END_AGE)) {
                    return false;
                }

                // 按照职业范围进行过滤
                if (!ValidUtils.in(aggrInfo, Const.SPARK.FIELD_PROFESSIONAL, parameter, Const.SPARK.PARAM_PROFESSIONALS)) {
                    return false;
                }
                // 按照城市范围进行过滤
                if (!ValidUtils.in(aggrInfo, Const.SPARK.FIELD_CITY, parameter, Const.SPARK.PARAM_CITIES)) {
                    return false;
                }
                // 按照性别进行过滤
                if (!ValidUtils.equal(aggrInfo, Const.SPARK.FIELD_SEX, parameter, Const.SPARK.PARAM_SEX)) {
                    return false;
                }
                // 按照搜索词进行过滤
                if (!ValidUtils.in(aggrInfo, Const.SPARK.FIELD_SEARCH_KEYWORDS, parameter, Const.SPARK.PARAM_KEYWORDS)) {
                    return false;
                }
                // 按照商品品类进行过滤
                if (!ValidUtils.in(aggrInfo, Const.SPARK.FIELD_CLICK_CATEGORY_IDS, parameter, Const.SPARK.PARAM_CATEGORY_IDS)) {
                    return false;
                }

                // if the program is executed here, explain that these sessions are needed,
                // we need to accumulate according to the session step and visit length

                sessionAggrStatAccumulator.add(Const.SESSION.SESSION_COUNT);

                long visitLength = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                        aggrInfo, "\\|", Const.SPARK.FIELD_VISIT_LENGTH));
                long stepLength = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                        aggrInfo, "\\|", Const.SPARK.FIELD_STEP_LENGTH));

                calculateVisitLength(visitLength);

                calculateStepLength(stepLength);

                return true;
            }

            private void calculateVisitLength(long visitLength) {
                if (visitLength >= 1 && visitLength <= 3) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_1s_3s);
                } else if (visitLength > 4 && visitLength <= 6) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_4s_6s);
                } else if (visitLength >= 7 && visitLength <= 9) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_7s_9s);
                } else if (visitLength >= 10 && visitLength < 29) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_10s_30s);
                } else if (visitLength >= 30 && visitLength < 60) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_30s_60s);
                } else if (visitLength >= 60 && visitLength < 180) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_1m_3m);
                } else if (visitLength >= 180 && visitLength < 600) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_3m_10m);
                } else if (visitLength >= 600 && visitLength < 1800) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_10m_30m);
                } else if (visitLength >= 1801) {
                    sessionAggrStatAccumulator.add(Const.SESSION.VISIT_LENGTH_30m);
                }
            }

            private void calculateStepLength(long stepLength) {
                if (stepLength >= 1 && stepLength <= 3) {
                    sessionAggrStatAccumulator.add(Const.SESSION.STEP_PERIOD_1_3);
                } else if (stepLength >= 4 && stepLength <= 6) {
                    sessionAggrStatAccumulator.add(Const.SESSION.STEP_PERIOD_4_6);
                } else if (stepLength >= 7 && stepLength <= 9) {
                    sessionAggrStatAccumulator.add(Const.SESSION.STEP_PERIOD_7_9);
                } else if (stepLength >= 10 && stepLength < 30) {
                    sessionAggrStatAccumulator.add(Const.SESSION.STEP_PERIOD_10_30);
                } else if (stepLength >= 30 && stepLength < 60) {
                    sessionAggrStatAccumulator.add(Const.SESSION.STEP_PERIOD_30_60);
                }
            }
        });
    }

    /**
     * calculate the proportion of session
     *
     * @param value  sessionAggrStatAccumulator
     * @param taskId task id
     */
    private static void calculateAndPersist(String value, long taskId) {
        long session_count = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.SESSION_COUNT));

        long visit_length_1s_3s = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_1s_3s));
        long visit_length_4s_6s = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_4s_6s));
        long visit_length_7s_9s = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_7s_9s));
        long visit_length_10s_30s = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_10s_30s));
        long visit_length_30s_60s = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_30s_60s));
        long visit_length_1m_3m = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_1m_3m));
        long visit_length_3m_10m = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_3m_10m));
        long visit_length_10m_30m = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_10m_30m));
        long visit_length_30m = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.VISIT_LENGTH_30m));

        long step_length_1_3 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_1_3));
        long step_length_4_6 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_4_6));
        long step_length_7_9 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_7_9));
        long step_length_10_30 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_10_30));
        long step_length_30_60 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_30_60));
        long step_length_60 = NumberUtils.toLong(StringUtils.getFieldFromConcatString(
                value, "\\|", Const.SESSION.STEP_PERIOD_60));

        double visit_length_1s_3s_ratio = NumberUtils.formatDouble((double) visit_length_1s_3s / session_count, 2);
        double visit_length_4s_6s_ratio = NumberUtils.formatDouble((double) visit_length_4s_6s / session_count, 2);
        double visit_length_7s_9s_ratio = NumberUtils.formatDouble((double) visit_length_7s_9s / session_count, 2);
        double visit_length_10s_30s_ratio = NumberUtils.formatDouble((double) visit_length_10s_30s / session_count, 2);
        double visit_length_30s_60s_ratio = NumberUtils.formatDouble((double) visit_length_30s_60s / session_count, 2);
        double visit_length_1m_3m_ratio = NumberUtils.formatDouble((double) visit_length_1m_3m / session_count, 2);
        double visit_length_3m_10m_ratio = NumberUtils.formatDouble((double) visit_length_3m_10m / session_count, 2);
        double visit_length_10m_30m_ratio = NumberUtils.formatDouble((double) visit_length_10m_30m / session_count, 2);
        double visit_length_30m_ratio = NumberUtils.formatDouble((double) visit_length_30m / session_count, 2);
        double step_length_1_3_ratio = NumberUtils.formatDouble((double) step_length_1_3 / session_count, 2);
        double step_length_4_6_ratio = NumberUtils.formatDouble((double) step_length_4_6 / session_count, 2);
        double step_length_7_9_ratio = NumberUtils.formatDouble((double) step_length_7_9 / session_count, 2);
        double step_length_10_30_ratio = NumberUtils.formatDouble((double) step_length_10_30 / session_count, 2);
        double step_length_30_60_ratio = NumberUtils.formatDouble((double) step_length_30_60 / session_count, 2);
        double step_length_60_ratio = NumberUtils.formatDouble((double) step_length_60 / session_count, 2);

        // package statistical results into a domain object

        SessionAggrStat sessionAggrStat = new SessionAggrStat();

        sessionAggrStat.setSessionCount(session_count);
        sessionAggrStat.setTaskId(taskId);
        sessionAggrStat.setVisit_length_1s_3s_ratio(visit_length_1s_3s_ratio);
        sessionAggrStat.setVisit_length_4s_6s_ratio(visit_length_4s_6s_ratio);
        sessionAggrStat.setVisit_length_7s_9s_ratio(visit_length_7s_9s_ratio);
        sessionAggrStat.setVisit_length_10s_30s_ratio(visit_length_10s_30s_ratio);
        sessionAggrStat.setVisit_length_30s_60s_ratio(visit_length_30s_60s_ratio);
        sessionAggrStat.setVisit_length_1m_3m_ratio(visit_length_1m_3m_ratio);
        sessionAggrStat.setVisit_length_3m_10m_ratio(visit_length_3m_10m_ratio);
        sessionAggrStat.setVisit_length_10m_30m_ratio(visit_length_10m_30m_ratio);
        sessionAggrStat.setVisit_length_30m_ratio(visit_length_30m_ratio);
        sessionAggrStat.setStep_length_1_3_ratio(step_length_1_3_ratio);
        sessionAggrStat.setStep_length_4_6_ratio(step_length_4_6_ratio);
        sessionAggrStat.setStep_length_7_9_ratio(step_length_7_9_ratio);
        sessionAggrStat.setStep_length_10_30_ratio(step_length_10_30_ratio);
        sessionAggrStat.setStep_length_30_60_ratio(step_length_30_60_ratio);
        sessionAggrStat.setStep_length_60_ratio(step_length_60_ratio);

        // insert
        ISessionAggrStatDAO sessionAggrStatDAO = DAOFactory.getSessionAggrStatDAO();
        sessionAggrStatDAO.insert(sessionAggrStat);
    }
}
